package cn.sheep.dmp.etl

import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Sheep.Old @ 64341393
  * Created 2018/3/28
  */
object Sql2Parquet1 {

  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setAppName("日志转parquet文件")
      .setMaster("local[*]")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") // RDD

    val sc = new SparkContext(sparkConf)
    val sqlc = new SQLContext(sc)

    //获取字段
    val parquet = sqlc.read.parquet("parquet")

    val cleanData = parquet.map(row => {

      val requestmode = row.getAs[Int]("requestmode") //原始请求数

      val processnode = row.getAs[Int]("processnode") //有效请求数

      val iseffective = row.getAs[Int]("iseffective") // 广告请求数

      val isbilling = row.getAs[Int]("isbilling") //参与竞价数

      val isbid = row.getAs[Int]("isbid") //竞价成功数

      val iswin = row.getAs[Int]("iswin") //展示数

      val adorderid = row.getAs[Int]("adorderid") //点击数
      //val adcreativeid = row.getAs[String]("adcreativeid")

      val winprice = row.getAs[Double]("winprice") //DSP广告消费

      val adpayment = row.getAs[Double]("adpayment")

      val provincename = row.getAs[String]("provincename")
      //DSP广告成本
      val cityname = row.getAs[String]("cityname")

      val firstReq = if (requestmode == 1 && processnode >= 1) 1 else 0
      if (requestmode == 1 && processnode >= 2) 1 else 0
      if (requestmode == 1 && processnode >= 3) 1 else 0
      if (iseffective == 1 && isbilling == 1 && isbid == 1 && adorderid != 1) 1 else 0
      if (iseffective == 1 && isbilling == 1 && iswin == 1) 1 else 0
      if (requestmode == 2 && iseffective == 1) 1 else 0
      if (requestmode == 3 && iseffective == 1) 1 else 0
      if (iseffective == 1 && isbilling == 1 && iswin == 1) winprice / 1000 else 0
      if (iseffective == 1 && isbilling == 1 && iswin == 1) adpayment / 1000 else 0

      // (provincename, cityname, requestmode, processnode, iseffective, isbilling, isbid, iswin, adorderid, winprice, adpayment)

      ((provincename, cityname), List(requestmode, processnode, iseffective, isbilling, isbid, iswin, adorderid, winprice, adpayment))


    })
    parquet.registerTempTable("t_log")

    /*  val udfFun = (b:Boolean) => if(b) 1 else 0

      sqlc.udf.register("udfFun",udfFun)*/

    sqlc.sql(
      """
        |select
        |
        |sum(case when requestmode = 1 and processnode >= 1 then 1 else 0 end) res,
        |sum(case when requestmode = 1 and processnode >= 2 then 1 else 0 end) effreq,
        |sum(case when requestmode = 1 and processnode = 3 then 1 else 0 end) adreq,
        |sum(case when iseffective = 1 and isbilling = 1 and isbid = 1 and adorderid != 0 then 1 else 0 end) jjreq,
        |sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 then 1 else 0 end) succreq,
        |sum(case when requestmode = 2 and iseffective = 1 then 1 else 0 end) showreq,
        |sum(case when requestmode = 3 and iseffective = 1 then 1 else 0 end) clickreq,
        |sum(case when iseffective = 1 and isbilling >= 1 and iswin = 1 then winprice/1000 else 0 end) ptices,
        |sum(case when iseffective = 1 and isbilling >= 1 and iswin = 1 then adpayment/1000 else 0 end) ptice

        |from t_log group by provincename,cityname
      """.stripMargin).show(10)

    /* cleanData.toDF("provincename", "cityname", "requestmode", "processnode", "iseffective", "isbilling", "isbid", "iswin", "adorderid", "winprice", "adpayment").registerTempTable("t_log")

      val dataFrame = sqlc.sql(
        """
            select provincename,cityname,sum(requestmode) requestmode,sum(processnode) processnode,sum(iseffective) iseffective,sum(isbilling) isbilling,sum(isbid) isbid,sum(iswin) iswin,sum(adorderid) adorderid,sum(winprice) winprice,sum(adpayment) adpayment
            from t_log group by provincename,cityname
        """.stripMargin)

      val prop = new Properties()
      val url = "jdbc:mysql://localhost:3306/aa?characterEncoding=utf-8"

      prop.setProperty("user", "root")
      prop.setProperty("password", "root123")
      dataFrame.write.jdbc(url, "area", prop)*/
    val value = cleanData.reduceByKey((tp1, tp2) => {


      tp1.zip(tp2).map(value => value._1 + value._2)
    })
    value.foreach(println)
    sc.stop()
  }

}



